/*
 * Copyright (c) 2011 Imre Fazekas.
 * All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * Neither the name of the Brillien nor the names of its
 * terms and concepts may be used to endorse or promote products derived from this
 * software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

package com.vii.brillien.ignition.transport.amqp;

import com.rabbitmq.client.*;
import com.vii.brillien.ignition.BrillienContext;
import com.vii.brillien.ignition.transport.BrillienCommunication;
import com.vii.brillien.ignition.transport.BrillienMediator;
import com.vii.brillien.ignition.transport.TransportStatics;
import com.vii.brillien.kernel.BrillienException;
import com.vii.brillien.kernel.axiom.transport.Communication;
import com.vii.streamline.services.ThreadServices;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;

/**
 * Mediator call handling all AMQP-related operations.
 */
public class AmqpMediator extends BrillienMediator<QueueingConsumer.Delivery> {

    /**
     * IP address of the AMQP server
     */
    protected String                    host;

    /**
     * Port of the AMQP server
     */
    protected int                       port;

    /**
     * Name and password of the user for authentication while logging in.
     */
    protected String                    username, password;

    protected String                    queuename;

    protected ConnectionFactory         factory;

    protected BlockingQueue<Connection> connQueue;

    protected AmqpListener              listener;

    protected boolean                   durable;

    {
        host = "localhost";
        port = AMQP.PROTOCOL.PORT;
        username = queuename = "liaison" + this.getClass().getSimpleName();
        password = "liaison";

        durable = false;

        connQueue = new LinkedBlockingQueue<Connection>(1);
    }

    protected class BrillienShutdownListener implements ShutdownListener {
        @Override
        public void shutdownCompleted(ShutdownSignalException e) {
            if( listener!=null && listener.isListen() ){
                connQueue.poll();
                try {
                    createConnection();
                } catch (BrillienException be) {
                    BrillienContext.exceptionLog( AmqpMediator.class.getName(), "shutdownCompleted", be );
                }
            }
        }
    }

    public AmqpMediator() {
    }

    public AmqpMediator(String host, int port, String queueName, boolean durable) {
        this(host, port, null, null, queueName, durable);
    }

    public AmqpMediator(String host, int port, String username, String password, String queuename, boolean durable) {
        this.host = host;
        this.port = port;
        this.username = username;
        this.password = password;

        this.queuename = queuename;

        this.durable = durable;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
        setQueuename( this.username.toLowerCase() );
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getQueuename() {
        return queuename;
    }

    public void setQueuename(String queuename) {
        this.queuename = queuename;
    }

    public boolean isDurable() {
        return durable;
    }

    public void setDurable(boolean durable) {
        this.durable = durable;
    }

    protected void createConnection() throws BrillienException {
        for( int i=0; i<BrillienContext.RECONNECTION_ATTEMPT; ++i ){
            try {
                Connection conn = factory.newConnection();

                conn.addShutdownListener(new BrillienShutdownListener());

                connQueue.offer(conn);

                return;
            } catch (IOException e) { }
            CountDownLatch latch = new CountDownLatch( 1 );
            try {
                latch.await( BrillienContext.RECONNECTION_TIMEOUT, TimeUnit.MILLISECONDS );
            } catch (InterruptedException e) { }
        }
        throw new BrillienException("Cannot get a connection");
    }
    protected Connection getConnection() throws BrillienException {
        Connection conn = null;
        try {
            for( int i=0; conn==null && i<BrillienContext.RECONNECTION_ATTEMPT; ++i ){
                BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Attempting to get a connection...", i );

                conn = connQueue.poll( BrillienContext.RECONNECTION_TIMEOUT, TimeUnit.MILLISECONDS );
            }
            if( conn == null )
                throw new BrillienException("Cannot get a connection");

            connQueue.offer( conn );
            return conn;
        } catch (InterruptedException e) {
            throw new BrillienException( e );
        }
    }

    @Override
    public void connect() throws BrillienException {
        BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Connecting to: " + host + " " + port + " " + queuename);

        factory = new ConnectionFactory();

        /*
        if (username != null)
            factory.setUsername(username);

        if (password != null)
            factory.setPassword(password);
        */
        factory.setHost(host);
        factory.setPort(port);

        createConnection();

        if (queuename == null )
            setQueuename(username);

        if (queuename == null)
            throw new BrillienException("QueueName hasn't been specified!");

        createQueue(getConnection(), queuename, isDurable(), false, false);

        BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Connected. " );

        startListening();
    }

    /**
     * Closes the connection with the XMPP server
     */
    @Override
    public void disconnect() throws BrillienException {
        try {
            listener.setListen(false);
            connQueue.peek().close();
        } catch (IOException e) {
            throw new BrillienException(e);
        }
        BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Disconnected.");
    }

    @Override
    public void startListening() throws BrillienException {
        BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Started listening...");

        ThreadServices.getStpe().submit(listener = new AmqpListener(this, queuename));
    }

    public void createQueue(Connection conn, String queueName) throws BrillienException {
        createQueue(conn, queueName, true, false, false);
    }

    public void createQueue(Connection conn, String queueName, boolean durable, boolean exclusive, boolean autoDelete) throws BrillienException {
        BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Creating queue " + queueName, queueName, durable, exclusive, autoDelete);

        Channel ch = null;
        try {
            ch = conn.createChannel();

            ch.queueDeclare(queueName, durable, exclusive, autoDelete, null);

            ch.close();

            BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + "Queue created.");
        } catch (IOException e) {
            throw new BrillienException(e);
        }
    }

    @Override
    public AmqpCommunication newCommunication() throws BrillienException {
        return new AmqpCommunication();
    }

    @Override
    public BrillienCommunication sendCommunication(long timeout, HashMap<String, Object> properties, final BrillienCommunication comm) throws BrillienException {
        try {
            Channel ch = getConnection().createChannel();

            AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder( );
            props.contentType(TransportStatics.TRANSPORT_MIME_TYPE);
            props.contentEncoding("UTF-8");
            props.deliveryMode(TransportStatics.MESSAGE_DELIVERY_PERSISTENT);
            props.correlationId(comm.getOriginalId());
            props.replyTo(this.getQueuename());
            props.messageId(comm.getId());
            props.timestamp(new Date(comm.getRequestTime()));
            props.type(comm.getMessageType() + "");
            props.appId( comm.getFlowID() );

            final LinkedBlockingQueue<BrillienCommunication> queue = new LinkedBlockingQueue<BrillienCommunication>(1);

            ch.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, java.lang.String replyText, java.lang.String exchange, java.lang.String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        BrillienCommunication comm_ = prepareNewCommunication(comm.getSender(), comm.getFlowID(), comm.getRecipient(), comm.getSubject(), Communication.TYPE_ERROR, null);
                        comm_.setErrorCode( replyCode ); comm_.setErrorValue( replyText );
                        queue.add( comm_ );
                    } catch (BrillienException e) {
                        BrillienContext.exceptionLog(AmqpMediator.class.getName(), "sendCommunication", e);
                    }
                }
            });

            if (comm.getMessageType() == Communication.TYPE_GET) {
                ResponseListener listener = new ResponseListener(comm, queue, this);

                addCommunicationListener(listener);
            }

            String messageToSend = getMessageToSend(comm);

            ch.basicPublish("", comm.getRecipient().toLowerCase(), props.build(), messageToSend.getBytes(Charset.forName("UTF-8")));

            BrillienContext.basicLog(Level.FINE, mediatorEntity() + ":: " + " at " + System.currentTimeMillis() + "Message sent " + messageToSend);

            BrillienCommunication response = null;

            if (comm.getMessageType() == Communication.TYPE_GET) {
                BrillienContext.basicLog( Level.FINE, mediatorEntity() + ":: " + "waiting for response for " + comm );

                response = queue.poll(timeout, TimeUnit.MILLISECONDS);

                BrillienContext.basicLog( Level.FINE, mediatorEntity() + ":: " + "response received " + response + " to " + comm );

                if (response == null) {
                    throw new BrillienException( "No response from server for communication: " + comm );
                }
                else if (response.getMessageType() == Communication.TYPE_ERROR) {
                    throw new BrillienException( response.getResponse().toString() );
                }
            }

            ch.close();

            return response;
        } catch (BrillienException e) {
            throw e;
        } catch (Exception e) {
            //BrillienContext.exceptionLog(AmqpMediator.class.getName(), "sendCommunication", e);
            throw new BrillienException(e);
        }
    }

    @Override
    public String mediatorEntity() {
        return queuename != null ? queuename : username;
    }

    @Override
    public String toString() {
        return "AmqpMediator{" +
                "host='" + host + '\'' +
                ", port=" + port +
                ", username='" + username + '\'' +
                ", password='" + password + '\'' +
                '}';
    }
}
